package org.easy.rocketmq;


import org.easy.rocketmq.dto.IMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.StringUtils;


@Slf4j
public class CommonRocketMqSender {

    private final RocketMQTemplate rocketMQTemplate;

    public CommonRocketMqSender(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    private Message buildMessage(IMessage message) {
        MessageBuilder<IMessage> iMessageMessageBuilder = MessageBuilder.withPayload(message);
        if (!StringUtils.isEmpty(message.getKey())) {
            iMessageMessageBuilder.setHeader(RocketMQHeaders.KEYS, message.getKey());
        }

        if (!StringUtils.isEmpty(message.getTag())) {
            iMessageMessageBuilder.setHeader(RocketMQHeaders.TAGS, message.getTag());
        }

        if (!StringUtils.isEmpty(message.getDelay())) {
            iMessageMessageBuilder.setHeader(RocketMQHeaders.DELAY, changeToDelayLevel(message.getDelay()));
        }

        Message<IMessage> build = iMessageMessageBuilder.build();

        return build;
    }

    /**
     * 发送消息
     *
     * @param context
     */
    public void send(String topic, IMessage context) {
        rocketMQTemplate.send(topic, buildMessage(context));
    }

    /**
     * 发送异步消息
     * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     *
     * @param delayLevel 延迟消息的级别
     */
    public void send(String topic, IMessage context, SendCallback sendCallback, int delayLevel) {
        rocketMQTemplate.asyncSend(topic, buildMessage(context), sendCallback, 20000, delayLevel);
    }

    /**
     * 发送单向消息
     *
     * @param context
     */
    public void sendOneWay(String topic, IMessage context) {
        rocketMQTemplate.sendOneWay(topic, buildMessage(context));
    }

    /**
     * 发送同步顺序消息
     *
     * @param hashKey 发送顺序消息时使用的hashKey主要用于确保相同业务标识的消息被发送到同一个队列中，从而保证消息的顺序性
     */
    public void sendOrderly(String topic, IMessage context, String hashKey) {
        log.info("发送顺序消息,topic:{},hashKey:{}", topic, hashKey);
        rocketMQTemplate.syncSendOrderly(topic, buildMessage(context), hashKey);
    }

    /**
     * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     *
     * @param
     * @return
     */
    private static int changeToDelayLevel(long delay) {
        if (delay <= 1000) {
            return 1;
        } else if (delay <= 5 * 1000) {
            return 2;
        } else if (delay <= 10 * 1000) {
            return 3;
        } else if (delay <= 30 * 1000) {
            return 4;
        } else if (delay <= 60 * 1000) {
            return 5;
        } else if (delay <= 2 * 60 * 1000) {
            return 6;
        } else if (delay <= 3 * 60 * 1000) {
            return 7;
        } else if (delay <= 4 * 60 * 1000) {
            return 8;
        } else if (delay <= 5 * 60 * 1000) {
            return 9;
        } else if (delay <= 6 * 60 * 1000) {
            return 10;
        } else if (delay <= 7 * 60 * 1000) {
            return 11;
        } else if (delay <= 8 * 60 * 1000) {
            return 12;
        } else if (delay <= 9 * 60 * 1000) {
            return 13;
        } else if (delay <= 10 * 60 * 1000) {
            return 14;
        } else if (delay <= 20 * 60 * 1000) {
            return 15;
        } else if (delay <= 30 * 60 * 1000) {
            return 16;
        } else if (delay <= 60 * 60 * 1000) {
            return 17;
        } else {
            return 18;
        }
    }



}
